# -*- coding: utf-8 -*-
from datetime import timedelta
from time import time
from utils.operators.rich_sql_sensor import RichSqlSensor
from jms.dm.dm_center_transfer_summary_dt import jms_dm__dm_center_transfer_summary_dt

label = f"dws_center_transfer_summary_dt_{int(time())}"
timeout = timedelta(minutes=30).seconds
doris_jms_dws__dws_center_transfer_summary_dt = RichSqlSensor(
    task_id='doris_jms_dws__dws_center_transfer_summary_dt',
    pool='broker_load_pool',
    email=['wangmenglei@jtexpress.com','yl_bigdata@yl-scm.com'],
    task_concurrency=1,
    conn_id='doris',
    pre_sql=f"""
	TRUNCATE TABLE jms_dws.dws_center_transfer_summary_dt PARTITION (
		 p{{{{ execution_date | cst_ds_nodash }}}}
		,p{{{{ execution_date | date_add(-1 ) | cst_ds_nodash }}}}
		,p{{{{ execution_date | date_add(-2 ) | cst_ds_nodash }}}}
	);
	LOAD LABEL jms_dws.{label} (
	    DATA INFILE(
	    	"hdfs://{{{{var.value.hadoop_namespace}}}}/dw/hive/jms_dm.db/external/dm_center_transfer_summary_dt/dt={{{{ execution_date | date_add(-2 ) | cst_ds }}}}/*",
	    	"hdfs://{{{{var.value.hadoop_namespace}}}}/dw/hive/jms_dm.db/external/dm_center_transfer_summary_dt/dt={{{{ execution_date | date_add(-1 ) | cst_ds }}}}/*",
	    	"hdfs://{{{{var.value.hadoop_namespace}}}}/dw/hive/jms_dm.db/external/dm_center_transfer_summary_dt/dt={{{{ execution_date | cst_ds }}}}/*"
	    )
	    INTO TABLE dws_center_transfer_summary_dt
	    FORMAT AS 'PARQUET'(
             scan_time           
            ,scan_site_code      
            ,scan_site_name      
            ,provider_id         
            ,provider_desc       
            ,city_id             
            ,city_desc           
            ,regional_id         
            ,regional_desc       
            ,agent_code          
            ,agent_name          
            ,order_source_code   
            ,order_source_name   
            ,oper_shift          
            ,arrival_cnt         
            ,send_cnt            
            ,arrival_pack_cnt    
            ,send_pack_cnt       
            ,enterport_cnt       
            ,same_city_cnt       
            ,transfer_cnt        
            ,leaveport_cnt       
            ,enterport_pack_cnt  
            ,same_city_pack_cnt  
            ,transfer_pack_cnt   
            ,leaveport_pack_cnt  
            ,enterport_weight_tot
            ,same_city_weight_tot
            ,transfer_weight_tot 
            ,leaveport_weight_tot
            ,trans_cnt           
            ,trans_pack_cnt      
            ,oper_cnt            
            ,oper_pack_cnt       
            ,oper_weight_tot    
            ,other_cnt
            ,other_pack_cnt
            ,other_weight_tot       
            ,sys_time         
            ,effect_start_time
            ,effect_end_time  
            ,shift_start_time 
            ,shift_end_time   
            ,sum_flag                        
	    )
	    columns from path as (dt)
	    set( 
            scan_time            =scan_time
           ,scan_site_code       =scan_site_code      
           ,scan_site_name       =scan_site_name      
           ,provider_id          =provider_id         
           ,provider_desc        =provider_desc       
           ,city_id              =city_id             
           ,city_desc            =city_desc           
           ,regional_id          =regional_id         
           ,regional_desc        =regional_desc       
           ,agent_code           =agent_code          
           ,agent_name           =agent_name          
           ,order_source_code    =order_source_code   
           ,order_source_name    =order_source_name   
           ,oper_shift           =oper_shift 
           ,arrival_cnt          =arrival_cnt         
           ,send_cnt             =send_cnt            
           ,arrival_pack_cnt     =arrival_pack_cnt    
           ,send_pack_cnt        =send_pack_cnt       
           ,enterport_cnt        =enterport_cnt       
           ,same_city_cnt        =same_city_cnt       
           ,transfer_cnt         =transfer_cnt        
           ,leaveport_cnt        =leaveport_cnt       
           ,enterport_pack_cnt   =enterport_pack_cnt  
           ,same_city_pack_cnt   =same_city_pack_cnt  
           ,transfer_pack_cnt    =transfer_pack_cnt   
           ,leaveport_pack_cnt   =leaveport_pack_cnt  
           ,enterport_weight_tot =enterport_weight_tot
           ,same_city_weight_tot =same_city_weight_tot
           ,transfer_weight_tot  =transfer_weight_tot 
           ,leaveport_weight_tot =leaveport_weight_tot
           ,trans_cnt            =trans_cnt           
           ,trans_pack_cnt       =trans_pack_cnt      
           ,oper_cnt             =oper_cnt            
           ,oper_pack_cnt        =oper_pack_cnt       
           ,oper_weight_tot      =oper_weight_tot   
           ,other_cnt            =other_cnt
           ,other_pack_cnt       =other_pack_cnt 
           ,other_weight_tot     =other_weight_tot
           ,sys_time             =sys_time
           ,effect_start_time    =effect_start_time
           ,effect_end_time      =effect_end_time
           ,shift_start_time     =shift_start_time
           ,shift_end_time       =shift_end_time  
           ,sum_flag             =sum_flag    
           ,dt                   =dt     
	    )
	) WITH BROKER '{{{{ var.json.doris_brokers | random_choice }}}}'
    PROPERTIES ('timeout'='{timeout}', 'max_filter_ratio'='0.0')""",
    poke_sql=f"SHOW LOAD FROM jms_dws WHERE label = '{label}' ORDER BY CreateTime DESC LIMIT 1",
    sql_on_kill=f"CANCEL LOAD FROM jms_dws WHERE LABEL = '{label}'",
    success=lambda r: r[2] == 'FINISHED',
    failure=lambda r: (r[2] is not None and r[2] == 'CANCELLED', str(r[7])),
    poke_interval=60,
    execution_timeout=timedelta(seconds=timeout + 120),
)

doris_jms_dws__dws_center_transfer_summary_dt << jms_dm__dm_center_transfer_summary_dt
